package at.punkt.sail.talis;

import at.punkt.repository.talis.TalisBooleanQuery;
import at.punkt.repository.talis.TalisGraphQuery;
import at.punkt.repository.talis.TalisTupleQuery;
import at.punkt.vocabulary.CHANGESET;
import info.aduna.iteration.CloseableIteration;
import info.aduna.iteration.CloseableIteratorIteration;
import info.aduna.iteration.ExceptionConvertingIteration;
import java.io.IOException;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Locale;
import java.util.StringTokenizer;
import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.UsernamePasswordCredentials;
import org.apache.commons.httpclient.auth.AuthScope;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.openrdf.model.BNode;
import org.openrdf.model.Graph;
import org.openrdf.model.Literal;
import org.openrdf.model.Namespace;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.GraphImpl;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.model.util.GraphUtil;
import org.openrdf.model.util.GraphUtilException;
import org.openrdf.model.vocabulary.RDF;
import org.openrdf.query.BindingSet;
import org.openrdf.query.BooleanQuery;
import org.openrdf.query.Dataset;
import org.openrdf.query.GraphQuery;
import org.openrdf.query.GraphQueryResult;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.Query;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.query.TupleQueryResultHandler;
import org.openrdf.query.TupleQueryResultHandlerException;
import org.openrdf.query.algebra.TupleExpr;
import org.openrdf.query.algebra.evaluation.TripleSource;
import org.openrdf.query.algebra.evaluation.impl.EvaluationStrategyImpl;
import org.openrdf.query.impl.EmptyBindingSet;
import org.openrdf.query.impl.GraphQueryResultImpl;
import org.openrdf.query.impl.TupleQueryResultBuilder;
import org.openrdf.query.resultio.QueryResultParseException;
import org.openrdf.query.resultio.sparqlxml.SPARQLBooleanXMLParser;
import org.openrdf.repository.RepositoryException;
import org.openrdf.rio.RDFHandler;
import org.openrdf.rio.RDFHandlerException;
import org.openrdf.rio.RDFParseException;
import org.openrdf.rio.helpers.StatementCollector;
import org.openrdf.rio.ntriples.NTriplesWriter;
import org.openrdf.rio.rdfxml.RDFXMLParser;
import org.openrdf.rio.rdfxml.RDFXMLWriter;
import org.openrdf.rio.rdfxml.util.RDFXMLPrettyWriter;
import org.openrdf.sail.SailConnection;
import org.openrdf.sail.SailException;
import org.openrdf.sail.helpers.NotifyingSailConnectionBase;

/**
 * Implementation of the {@link SailConnection} interface for the Talis metabox.<br />
 * Queries as well as addition and removal of statements are posted to the
 * Talis web services using an HttpClient.<br />
 * <ul>
 * <li>Read operations are translated to SPARQL queries and posted to the Talis SPARQL service</li>
 * <li>Create and delete operations are converted into a changeset and posted to the Talis RDF sink</li>
 * <li>Namespaces are currently not supported</li>
 * </ul>
 *
 * @author Alexander Kreiser
 * @author Jürgen Jakobitsch
 */
public class TalisSailConnection extends NotifyingSailConnectionBase {

    protected final Logger _logger = Logger.getLogger(TalisSailConnection.class.getName());

    // Parent TalisSail object
    protected final TalisSail _sail;

    // Cache for storing changes (add and remove statements) in a RDF changeset
    protected Graph _changesetCache;
    
    // Cache for storing statement additions as raw RDF (without changeset)
    protected Graph _addRDFCache;

    // RDF/XML writer for changesets and raw RDF data
    protected RDFXMLWriter _writer;

    // Format for RDF dates
    protected final static DateFormat iso8601Format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ", Locale.US);

    // HttpClient that will handle communication with the talis server
    protected final HttpClient _client;

    // Whether new RDF data should be added via changeset.
    // If this is set to false, the RDF data will be posted as raw RDF
    // to the RDF sink of the talis server.
    protected boolean _doChangesetAdd = true;

    protected class TalisTripleSource implements TripleSource {

        protected final ValueFactory _factory = new ValueFactoryImpl();
        protected final SailConnection _sailCon;
        protected final boolean _includeInferred;

        public TalisTripleSource(SailConnection sailCon, boolean includeInferred) {
            _includeInferred = includeInferred;
            _sailCon = sailCon;
        }

        public CloseableIteration<? extends Statement, QueryEvaluationException> getStatements(Resource rsrc, URI uri, Value value, Resource... rsrcs) throws QueryEvaluationException {
            try {
                return new ExceptionConvertingIteration<Statement, QueryEvaluationException>(_sailCon.getStatements(rsrc, uri, value, _includeInferred, rsrcs)) {

                    @Override
                    protected QueryEvaluationException convert(Exception excptn) {
                        return new QueryEvaluationException(excptn);
                    }
                };
            } catch (SailException e) {
                throw new QueryEvaluationException(e.getMessage(), e);
            }
        }

        public ValueFactory getValueFactory() {
            return _factory;
        }

    }

    public TalisSailConnection(TalisSail repository, HttpClient client) {
        super(repository);
        _sail = repository;
        _client = client;
        _changesetCache = new GraphImpl();
        _addRDFCache = new GraphImpl();
    }

    /**
     * Sets whether new RDF data should be added via changeset.
     * If this is set to false, the RDF data will be posted as raw RDF
     * to the RDF sink of the talis server.
     *
     * @param doChangesetAdd
     */
    public void setChangesetAdd(boolean doChangesetAdd) {
        _doChangesetAdd = doChangesetAdd;
    }

    /**
     * Add a statement that should be removed to the changeset in the cache.
     *
     * @param subject Subject of the statement
     * @param predicate Predicate of the statement
     * @param object Object of the statement
     */
    private void addRemoveStatementToCache(Resource subject, URI predicate, Value object) {
        Resource currentChangeset = getChangeSetForSubject(subject);
        BNode currentRemoval = this.getValueFactory().createBNode();
        _changesetCache.add(currentChangeset, CHANGESET.REMOVAL, currentRemoval);
        _changesetCache.add(currentRemoval, RDF.TYPE, RDF.STATEMENT);
        _changesetCache.add(currentRemoval, RDF.SUBJECT, subject);
        _changesetCache.add(currentRemoval, RDF.PREDICATE, predicate);
        _changesetCache.add(currentRemoval, RDF.OBJECT, object);
    }

    /**
     * Returns the changeset for a given subject from the cache.
     * If no changeset exists yet a new one will be created and returned.
     *
     * @param subject The subject of the changeset (cs:subjectOfChange)
     * @return The Resource representing the changeset
     */
    private Resource getChangeSetForSubject(Resource subject) {
        // @TODO Rewrite this to run without GraphUtil
        try {
            Resource currentChangeset = GraphUtil.getOptionalSubject(_changesetCache, CHANGESET.SUBJECTOFCHANGE, subject);
            if (currentChangeset == null) {
                // There is no changeset for this subject yet!
                // Create a new one
                currentChangeset = this.getValueFactory().createURI("urn:uuid:"+UUID.randomUUID().toString());
                _changesetCache.add(currentChangeset, RDF.TYPE, CHANGESET.CHANGESET);
                _changesetCache.add(currentChangeset, CHANGESET.SUBJECTOFCHANGE, subject);
                _changesetCache.add(currentChangeset, CHANGESET.CHANGEREASON, this.getValueFactory().createLiteral("das dao"));
                _changesetCache.add(currentChangeset, CHANGESET.CREATORNAME, this.getValueFactory().createLiteral(((UsernamePasswordCredentials) _client.getState().getCredentials(AuthScope.ANY)).getUserName()));
                _changesetCache.add(currentChangeset, CHANGESET.CREATEDDATE, this.getValueFactory().createLiteral(iso8601Format.format(new Date())));
            }
            return currentChangeset;
        } catch (GraphUtilException ex) {
            _logger.log(Level.SEVERE, null, ex);
            return null;
        }        
    }

    public Query prepareQuery(QueryLanguage language, String query, String baseURI) throws RepositoryException, MalformedQueryException {
        StringTokenizer st = new StringTokenizer(query);
        String type = st.nextToken().toLowerCase();
        if(type.equals("select")) {
                return prepareTupleQuery(language, query, baseURI);
        }
        else if(type.equals("construct") || type.equals("describe")) {
                return prepareGraphQuery(language, query, baseURI);
        }
        else if(type.equals("ask")) {
                return prepareBooleanQuery(language, query, baseURI);
        }
        else {
            return null;
        }
    }

    public TupleQuery prepareTupleQuery(QueryLanguage ql, final String query, String baseURI) throws RepositoryException, MalformedQueryException {

        TupleQuery q = new TalisTupleQuery() {
            @Override
            public TupleQueryResult evaluate() throws QueryEvaluationException {
                final TupleQueryResultBuilder resultBuilder = new TupleQueryResultBuilder();
                executeSPARQLForHandler(query, resultBuilder);
                return resultBuilder.getQueryResult();
            }

            @Override
            public void evaluate(TupleQueryResultHandler handler) throws QueryEvaluationException, TupleQueryResultHandlerException {
                executeSPARQLForHandler(query, handler);
            }
        };
        return q;
    }
    
    /**
     * Post a new SPARQL query to the Talis SPARQL service
     * and return the PostMethod so its reponse body can be processed.
     * 
     * @param query The query as string
     * @return PostMethod containing the actual response
     * @throws UnsupportedEncodingException The encoding for the query went wrong
     * @throws IOException Something went wrong while (http) posting the query to the SPARQL service
     */
    private PostMethod postSPARQLQuery(String query) throws UnsupportedEncodingException, IOException {
        _logger.log(Level.INFO, query);
        PostMethod post = new PostMethod(_sail.getRepositoryUrl()+"/services/sparql");
        String entityBody = "query=" + URLEncoder.encode(query, "UTF-8");
        post.setRequestEntity(new StringRequestEntity(entityBody, "application/x-www-form-urlencoded", "UTF-8"));
        post.addRequestHeader("Content-Type", "application/x-www-form-urlencoded");
        post.addRequestHeader("Accept", "*/*");
        int code = _client.executeMethod(post);
        if (code != 200) {
            throw new IOException("The HTTP Post returned status code: "+code);
        }
        return post;
    }

    /**
     * Executes a SPARQL query, the response will be parsed and handled by the given handler.
     *
     * @param query SPARQL query as string
     * @param handler A handler for a TupleQueryResult that handles the response
     */
    protected void executeSPARQLForHandler(String query, TupleQueryResultHandler handler) {
        try {
            PostMethod post = postSPARQLQuery(query);
            _sail.getSPARQLResultsXMLParser().setTupleQueryResultHandler(handler);
            _sail.getSPARQLResultsXMLParser().parse(post.getResponseBodyAsStream());
        } catch (QueryResultParseException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (TupleQueryResultHandlerException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            _logger.log(Level.SEVERE, null, ex);
        }
    }

    public GraphQuery prepareGraphQuery(QueryLanguage ql, final String query, String baseURI) throws RepositoryException, MalformedQueryException {
        TalisGraphQuery q = new TalisGraphQuery() {
            
            @Override
            public GraphQueryResult evaluate() throws QueryEvaluationException {
                return executeSPARQLForGraphResult(query);
            }

            @Override
            public void evaluate(RDFHandler handler) throws QueryEvaluationException, RDFHandlerException {
                executeSPARQLForHandler(query, handler);
            }

        };
        return q;
    }

    /**
     * Executes a SPARQL query, the response will be parsed and handled by the given handler.
     *
     * @param query SPARQL query as string
     * @param handler A handler for a TupleQueryResult that handles the response
     */
    protected void executeSPARQLForHandler(String query, RDFHandler handler) {
        try {
            PostMethod post = postSPARQLQuery(query);
            RDFXMLParser parser = new RDFXMLParser();
            parser.setRDFHandler(handler);
            parser.parse(post.getResponseBodyAsStream(), "");
        } catch (RDFParseException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (RDFHandlerException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            _logger.log(Level.SEVERE, null, ex);
        }

    }

    /**
     * Executes a SPARQL query and returns the response as a GraphQueryResult.
     *
     * @param query SPARQL query as string.
     * @return response formatted as a GraphQueryResult
     */
    protected GraphQueryResult executeSPARQLForGraphResult(String query) {
        try {
            PostMethod post = postSPARQLQuery(query);
            RDFXMLParser parser = new RDFXMLParser();
            StatementCollector collector = new StatementCollector();
            parser.setRDFHandler(collector);
            parser.parse(post.getResponseBodyAsStream(), "");

            GraphQueryResult result = new GraphQueryResultImpl(new HashMap<String, String>(), collector.getStatements());
            return result;
        } catch (RDFParseException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (RDFHandlerException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (UnsupportedEncodingException ex) {
            _logger.log(Level.SEVERE, null, ex);
        } catch (IOException ex) {
            _logger.log(Level.SEVERE, null, ex);
        }
        return null;
    }

    public BooleanQuery prepareBooleanQuery(QueryLanguage ql, final String query, String baseURI) throws RepositoryException, MalformedQueryException {
        TalisBooleanQuery q = new TalisBooleanQuery() {
            @Override
            public boolean evaluate() throws QueryEvaluationException {
                try {
                    return executeSPARQLForBoolean(query);
                } catch (Exception ex) {
                    throw new QueryEvaluationException(ex);
                }
            }
        };
        return q;
    }

    private boolean executeSPARQLForBoolean(String query) throws UnsupportedEncodingException, IOException, QueryResultParseException {
         PostMethod post = postSPARQLQuery(query);
         SPARQLBooleanXMLParser parser = new SPARQLBooleanXMLParser();
         return parser.parse(post.getResponseBodyAsStream());
    }

    /**
     * Builds and returns a SPARQL query from a given triple pattern.
     *
     * @param subject The triple pattern subject, can be null
     * @param predicate The triple pattern predicate, can be null
     * @param object The triple pattern object, can be null
     * @param contexts Not supported yet
     * @return The built SPARQL query as strings
     */
    private String buildTriplePatternQuery(Resource subject, URI predicate, Value object) {
        String query = "SELECT ?s ?p ?o ";

        query += "WHERE { ?s ?p ?o. ";
        if (subject != null)
        {
           query += " FILTER ( ?s = ";
           if (subject instanceof URI)
               query += "<"+subject.stringValue()+"> ";
           else
               query += subject.stringValue()+" ";
           query += "). ";
        }
        if (predicate != null)
           query += " FILTER ( ?p = <"+predicate.toString()+"> ). ";
        if (object != null) {
           query += " FILTER ( ?o = ";
           if (object instanceof URI) {
               query += "<"+((URI)object).toString()+"> ";
           } else if (object instanceof BNode) {
               query += ((BNode)object).toString()+" ";
           } else if (object instanceof Literal) {
               query += ((Literal)object).toString()+" ";
           }
           query += "). ";
        }
        query += " }";

        return query;
    }

    public void exportStatements(Resource subject, URI predicate, Value object, boolean includeInferred, RDFHandler handler, Resource... contexts) throws RepositoryException, RDFHandlerException {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    /**
     * Posts all changesets that are currently in the cache to the Talis changeset service.
     *
     * @return The post method
     * @throws RDFHandlerException
     * @throws UnsupportedEncodingException
     * @throws IOException
     */
    private void postChangeSet() throws RDFHandlerException, UnsupportedEncodingException, IOException {
        PostMethod post = new PostMethod(_sail.getRepositoryUrl()+"/meta/changesets");
        StringWriter stringWriter = new StringWriter();
        _writer = new RDFXMLPrettyWriter(stringWriter);
        _writer.startRDF();
        for (Statement stmt : _changesetCache) {
            _writer.handleStatement(stmt);
        }
        _writer.endRDF();
        post.setRequestEntity(new StringRequestEntity(stringWriter.toString(), "application/vnd.talis.changeset+xml", "UTF-8"));
        System.out.println(stringWriter.toString());
        post.addRequestHeader("Content-Type", "application/vnd.talis.changeset+xml");
        stringWriter.close();
        int code = _client.executeMethod(post);
        if (code != 200 && code != 202 && code != 201) {
            throw new IOException("The HTTP Post returned status code: "+code);
        }
    }

    private void postRDF() throws RDFHandlerException, UnsupportedEncodingException, IOException {
        PostMethod post = new PostMethod(_sail.getRepositoryUrl()+"/meta/changesets");
        StringWriter stringWriter = new StringWriter();
        NTriplesWriter writer = new NTriplesWriter(stringWriter);
        writer.startRDF();
        for (Statement stmt : _addRDFCache) {
            writer.handleStatement(stmt);
        }
        writer.endRDF();
        post.setRequestEntity(new StringRequestEntity(stringWriter.toString(), "text/turtle", "UTF-8"));
        System.out.println(stringWriter.toString());
        post.addRequestHeader("Content-Type", "text/turtle");
        stringWriter.close();
        int code = _client.executeMethod(post);
        if (code != 204 && code != 201) {
            throw new IOException("The HTTP Post returned status code: "+code);
        }
    }

    @Override
    protected void closeInternal() throws SailException {
    }

    @Override
    protected CloseableIteration<? extends BindingSet, QueryEvaluationException> evaluateInternal(TupleExpr tupleExpr, Dataset dataset, BindingSet bindings, boolean includeInferred) throws SailException {
        logger.trace("Incoming query model:\n{}", tupleExpr.toString());
        try {

            TripleSource tripleSource = new TalisTripleSource(this, includeInferred);
            EvaluationStrategyImpl strategy = new EvaluationStrategyImpl(tripleSource, dataset);

            return strategy.evaluate(tupleExpr, EmptyBindingSet.getInstance());
        }
        catch (QueryEvaluationException e) {
            throw new SailException(e);
        }
        catch (RuntimeException e) {
            throw e;
        }

    }

    @Override
    protected CloseableIteration<? extends Resource, SailException> getContextIDsInternal() throws SailException {
        throw new UnsupportedOperationException("Talis REST API currently does not support namespaces.");
    }

    @Override
    protected CloseableIteration<? extends Statement, SailException> getStatementsInternal(Resource subj, URI pred, Value obj, boolean includeInferred, Resource... contexts) throws SailException {
        try {
            String query = buildTriplePatternQuery(subj, pred, obj);
            Collection<Statement> statements = new LinkedList<Statement>();
            TupleQuery q = prepareTupleQuery(QueryLanguage.SPARQL, query, "");
            TupleQueryResult result = q.evaluate();
            ValueFactory factory = getValueFactory();
            Resource s;
            URI p;
            Value o;
            while (result.hasNext()) {
                BindingSet bSet = result.next();
                s = (Resource) bSet.getBinding("s").getValue();
                p = (URI) bSet.getBinding("p").getValue();
                o = bSet.getBinding("o").getValue();
                statements.add(factory.createStatement(s, p, o));
            }
            return new CloseableIteratorIteration(statements.iterator());
        } catch (Exception ex) {
            _logger.log(Level.SEVERE, null, ex);
            throw new SailException(ex);
        }
    }

    @Override
    protected long sizeInternal(Resource... contexts) throws SailException {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    @Override
    protected void startTransactionInternal() throws SailException {
        
    }

    @Override
    protected void commitInternal() throws SailException {
        try {
            postChangeSet();
            if (!_doChangesetAdd) {
                postRDF();
            }
            rollbackInternal();
        } catch (Exception ex) {
            logger.error(ex.toString());
            throw new SailException(ex);
        }
    }

    @Override
    protected void rollbackInternal() throws SailException {
        _changesetCache.clear();
        _addRDFCache.clear();
    }

    @Override
    protected void addStatementInternal(Resource subj, URI pred, Value obj, Resource... contexts) throws SailException {
        if (_doChangesetAdd) {
            // Get the changeset for this subject in the cache
            Resource currentChangeset = getChangeSetForSubject(subj);

            // Create a new blank node where statement to add goes into
            Resource currentAddition = this.getValueFactory().createBNode();
            // Add the blank node and the actual statement to the changeset in the cache
            _changesetCache.add(currentChangeset, CHANGESET.ADDITION, currentAddition);
            _changesetCache.add(currentAddition, RDF.TYPE, RDF.STATEMENT);
            _changesetCache.add(currentAddition, RDF.SUBJECT, subj);
            _changesetCache.add(currentAddition, RDF.PREDICATE, pred);
            _changesetCache.add(currentAddition, RDF.OBJECT, obj);
        } else {
            _addRDFCache.add(subj, pred, obj);
        }
        notifyStatementAdded(getValueFactory().createStatement(subj, pred, obj));
    }

    @Override
    protected void removeStatementsInternal(Resource subj, URI pred, Value obj, Resource... contexts) throws SailException {
        if (subj == null || pred == null || obj == null) {
            // If either predicate or object is null, it means we have a triple pattern
            // We need to fetch the affected triples
            CloseableIteration<? extends Statement, SailException> statements = getStatements(subj, pred, obj, false);
            while (statements.hasNext()) {
                Statement s = statements.next();
                addRemoveStatementToCache(s.getSubject(), s.getPredicate(), s.getObject());
            }
        } else {
            // If all three parameters are set, we can just add the statement to remove to the cache
            addRemoveStatementToCache(subj, pred, obj);
        }
    }

    @Override
    protected void clearInternal(Resource... contexts) throws SailException {
    }

    @Override
    protected CloseableIteration<? extends Namespace, SailException> getNamespacesInternal() throws SailException {
        throw new SailException("Talis REST API currently does not support namespaces.");
    }

    @Override
    protected String getNamespaceInternal(String prefix) throws SailException {
        throw new SailException("Talis REST API currently does not support namespaces.");
    }

    @Override
    protected void setNamespaceInternal(String prefix, String name) throws SailException {
        throw new SailException("Talis REST API currently does not support namespaces.");
    }

    @Override
    protected void removeNamespaceInternal(String prefix) throws SailException {
        throw new SailException("Talis REST API currently does not support namespaces.");
    }

    @Override
    protected void clearNamespacesInternal() throws SailException {
        throw new SailException("Talis REST API currently does not support namespaces.");
    }

    private ValueFactory getValueFactory() {
        return _sail.getValueFactory();
    }
}
